package net.shambolica.jport.runner;

import net.erl.objects.ErlTerm;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.net.ProtocolException;

/**
 * A simple connection implemented over a pair of Input+Output streams.
 * Each term is preceded by a 4-byte word (big-endian) containing the number of
 * bytes in the term. A length of 0 or less is considered INVALID and will cause the connection
 * to be stopped/closed.
 * The reason for this to be abstract is that at time of creation it is unclear if
 * clean ETF is all I need.
 */
public abstract class SimpleStreamConnection implements DataConnection {

    protected final Logger log = LoggerFactory.getLogger(SimpleStreamConnection.class);

    private final InputStream input;
    private final OutputStream output;
    private volatile boolean isStopped;
    private static final int MAX_TERM_LENGTH = 16*1024*1024 + 1; // 16M.

    /**
     * Constructor over reader and writer.
     * The state will be set to CONNECTED, since the handshake is yet to be
     * performed.
     * @param input The Reader for input.
     * @param output The Writer for output.
     * @param isStopped The run/stop state of the connection, out of the gate.
     */
    public SimpleStreamConnection(InputStream input, OutputStream output, boolean isStopped) {
        this.input = input;
        this.output = output;
        this.isStopped = isStopped;
    }


    /**
     * Implement in subclasses: deserialize a term from a byte buffer.
     * @param buf   Buffer to deserialize from.
     * @return      A term, or null if none found.
     */
    public abstract ErlTerm parseTerm(byte[] buf) throws IOException;

    /**
     * Implement in subclasses: serialize a term to a byte buffer.
     * @param term  Term to serialize.
     * @return      Representing byte array.
     */
    public abstract byte[] serializeTerm(ErlTerm term) throws IOException;


    public synchronized void close() throws IOException {
        if (! isStopped) {
            if (input != null) input.close();
            if (output != null) output.close();
            isStopped = true;
        }
    }

    public boolean isClosed() {
        return isStopped;
    }

    public synchronized ErlTerm readTerm()
            throws InterruptedException, IOException {
        try {
            if (isClosed()) {
                return null;
            } else {
                byte[] lenbuf = readBuf(input, 4);
                int termlen = (((int)lenbuf[0] & 0xff) << 24)
                        | (((int)lenbuf[1] & 0xff) << 16)
                        | (((int)lenbuf[2] & 0xff) << 8) 
                        | ((int)lenbuf[3] & 0xff);
                if (termlen < 0) {
                    log.error("protocol error: negative term length ({})", termlen);
                    throw new ProtocolException("negative term length");
                }
                if (termlen > MAX_TERM_LENGTH) {
                    log.error("protocol error: term too long ({})", termlen);
                    throw new ProtocolException("term too long");
                }
                byte[] rdbuf = readBuf(input, termlen);
                ErlTerm ret = parseTerm(rdbuf);
                log.debug("got {}", ret);
                return ret;
            }

        } catch (IOException e) {
            log.info("IO Exception ({}: {}): setting connection state to STOPPED", e.getClass().getName(), e.getMessage());
            isStopped = true;
            throw e;
        } catch (RuntimeException e) {
            log.info("Runtime Exception ({}: {}): setting connection state to STOPPED", e.getClass().getName(), e.getMessage());
            isStopped = true;
            throw e;
        }
    }

    public synchronized void writeTerm(ErlTerm term)
            throws InterruptedException, IOException {
        try {
            if (! isClosed()) {
                byte[] termbuf = serializeTerm(term);
                int len = termbuf.length;
                byte[] lenbuf = new byte[]{
                        (byte)((len >> 24) & 0xff),
                        (byte)((len >> 16) & 0xff),
                        (byte)((len >> 8) & 0xff),
                        (byte)(len & 0xff)};
                writeBuf(output, lenbuf);
                writeBuf(output, termbuf);
            }
        } catch (IOException e) {
            log.info("IO Exception ({}: {}): setting connection state to STOPPED", e.getClass().getName(), e.getMessage());
            isStopped = true;
            throw e;
        } catch (RuntimeException e) {
            log.info("Runtime Exception ({}: {}): setting connection state to STOPPED", e.getClass().getName(), e.getMessage());
            isStopped = true;
            throw e;
        }
    }


    /**
     * Read exactly <i>len</i> bytes from <i>input</i>.
     * Note that if len == 0, a buf[0] is immediately returned, and no read operation
     * is actually performed.
     * @param input The input stream.
     * @param len   The exact number of bytes to read.
     * @return A byte buffer containing the bytes read.
     * @throws IOException if a failure to read occurs for some reason..
     */
    public static byte[] readBuf(InputStream input, int len) throws IOException {
        if (len < 0) {
            throw new IllegalArgumentException("negative length");
        }

        byte[] buf = new byte[len];
        if (len > 0) {
            for (int offset = 0 ; offset < buf.length ; ) {
                int nRead = input.read(buf, offset, buf.length - offset);
                if (nRead < 0) {
                    throw new ProtocolException("premature end of input stream");
                }
                offset += nRead;
            }
        }

        return buf;
    }


    /**
     * Write out the buffer <i>buf</i> to <i>output</i>.
     * If the buffer is null or empty (length == 0), return immediately, performing
     * no operations on the stream.
     * @param output The stream to which we are writing.
     * @param buf The byte buffer being output.
     * @throws java.io.IOException if a write error occurs.
     */
    public static void writeBuf(OutputStream output, byte[] buf)
            throws IOException {
        if (buf != null && buf.length > 0) {
            output.write(buf);
        }
    }

}
